package mq.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.toolkit.CollectionUtils;
import lombok.extern.slf4j.Slf4j;
import mq.dao.mapper.MqMessageAbnormalMapper;
import mq.domain.MqMessageAbnormal;
import mq.service.IMqMessageAbnormalService;
import mq.util.MqUtil;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;

/**
 * @description: mq消费者到交换机，交换机到队列异常消息记录(MqMessageAbnormal)表服务实现类
 * @author: zk
 * @since: 2022-08-22 13:58:07
 */
@Slf4j
@Service
public class MqMessageAbnormalServiceImpl implements IMqMessageAbnormalService {
    @Resource
    private MqMessageAbnormalMapper mqMessageAbnormalMapper;
    @Resource
    private RabbitTemplate rabbitTemplate;
    @Resource
    private MqUtil mqUtil;

    @Override
    public void resendMessage(Integer type) {
        log.info("MQ异常消息重新发送 start,type={}", type);
        Long preId = null;
        List<MqMessageAbnormal> mqMessageAbnormalList = mqMessageAbnormalMapper.queryByStatus(preId, MqMessageAbnormal.STATUS_FAIL, type, 5000);
        while (CollectionUtils.isNotEmpty(mqMessageAbnormalList)) {
            preId = mqMessageAbnormalList.get(mqMessageAbnormalList.size() - 1).getId();
            MqMessageAbnormal mqMessageAbnormal = null;
            List<Long> failedIds = new ArrayList<>();
            for (int i = 0; i < mqMessageAbnormalList.size(); i++) {
                mqMessageAbnormal = mqMessageAbnormalList.get(i);
                try {
                    // 先更新状态
                    MqMessageAbnormal abnormal = new MqMessageAbnormal();
                    abnormal.setId(mqMessageAbnormal.getId());
                    abnormal.setStatus(MqMessageAbnormal.STATUS_SUCCESS);
                    abnormal.setUpdateTime(new Date());
                    mqMessageAbnormalMapper.updateById(abnormal);
                    // 重新发送
                    reSend(mqMessageAbnormal);
                } catch (Exception e) {
                    log.error("重新发送失败 mqMessageAbnormal={}", mqMessageAbnormal, e);
                    failedIds.add(mqMessageAbnormal.getId());
                }
            }
            // 批量更新为发送失败
            if (CollectionUtils.isNotEmpty(failedIds)) {
                mqMessageAbnormalMapper.updateStatusByIds(failedIds, MqMessageAbnormal.STATUS_FAIL);
                log.info("重新发送失败条数 size={}", failedIds.size());
            }
            mqMessageAbnormalList = mqMessageAbnormalMapper.queryByStatus(preId, MqMessageAbnormal.STATUS_SUCCESS, type, 5000);
        }
        log.info("MQ异常消息重新发送 end");
    }

    /**
     * 发送消息到MQ中
     *
     * @param mqMessageAbnormal
     */
    private void reSend(MqMessageAbnormal mqMessageAbnormal) {
        if (Objects.isNull(mqMessageAbnormal)) {
            log.warn("MQ消息mqMessageAbnormal为空!");
            return;
        }
        Message message = JSON.parseObject(mqMessageAbnormal.getMessage(), Message.class);
        // correlationData
        CorrelationData correlationData = new CorrelationData();
        MqMessageAbnormal correlationAbnormal = new MqMessageAbnormal();
        correlationAbnormal.setExchange(mqMessageAbnormal.getExchange());
        correlationAbnormal.setRoutingKey(mqMessageAbnormal.getRoutingKey());
        correlationAbnormal.setMessage(mqMessageAbnormal.getMessage());
        Message correlationMessage = mqUtil.getMessage(mqMessageAbnormal.getMessageId(), mqMessageAbnormal);
        correlationData.setReturnedMessage(correlationMessage);
        // send
        rabbitTemplate.convertAndSend(mqMessageAbnormal.getExchange(), mqMessageAbnormal.getRoutingKey(), message, correlationData);
    }

}
